package com.shujia.spark.core

import org.apache.spark.{Partitioner, SparkConf, SparkContext}

object Demo23Repartition {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("app")

    //设置sparksql  默认分区数，   在产生shuffle过程中会起作用,shuffle之后rdd分区的数量
    //conf.set("spark.sql.shuffle.partitions", "1")
    conf.set("spark.default.parallelism", "2")

    val sc = new SparkContext(conf)

    /**
      * 当读取hdfs文件的时候，rdd分区数由block数量决定，切片规则和mr一样
      *
      * minPartitions:  分区数据最小的限制,  rdd分区数不会比这个值小
      *
      * rdd分区数
      * 默认切片数 和 minPartitions 取最大值
      *
      */
    val rdd1 = sc.textFile("spark/data/stage", 4)

    println(s"rdd1的分区数：${rdd1.getNumPartitions}")

    //不产生shuffle的算子不能改变rdd的分区数
    val rdd2 = rdd1.flatMap(_.split(",")).map((_, 1))
    println(s"rdd2的分区数：${rdd2.getNumPartitions}")


    /**
      * 所有会产生shuffle的算子都可以指定一个numPartitions的参数
      *
      * numPartitions：shuffle之后rdd分区的数量（reduce的数量）
      * numPartitions不指定，默认shuffle之后rdd的分区数量和shuffle之前一致
      *
      * conf.set("spark.default.parallelism", "2") 全局默认分区数
      *
      * numPartitions优先级比spark.default.parallelism优先级高
      *
      * rdd2.groupByKey(numPartitions = 2)
      * rdd2.reduceByKey((x, y) => x + y, 2)
      */
    val rdd3 = rdd2.reduceByKey((x, y) => x + y, 3)


    println(s"rdd3的分区数：${rdd3.getNumPartitions}")


    /**
      * 重分区：增加或者减少并行度
      *
      * repartition  :  可以用于增加分区和减少分区，会产生shuffle
      * coalesce  :  有shuffle的条件是既可以用于增加分区也可以用于减少分区，没有shuffle的条件是只能用于减少分区
      *
      */

    val rdd4 = rdd3.repartition(5)
    println(s"rdd4的分区数：${rdd4.getNumPartitions}")


    val rdd5 = rdd4.coalesce(6, false)
    println(s"rdd5的分区数：${rdd5.getNumPartitions}")


    /**
      * 自定义分区
      *
      */

    val student = sc.textFile("spark/data/students.txt")

    val restudent = student
      .map(line => (line.split(",")(2).toInt, line))
      .groupByKey(new MyPartition)

    restudent.mapPartitionsWithIndex((partition, iter) => {

      println("当前分区编号:" + partition)
      println("当前分区数据:" + iter.toList)

      iter
    })
      //触发任务执行
      .foreach(x => x)


  }
}


class MyPartition extends Partitioner {
  //获取分区的数量
  override def numPartitions: Int = 4// 0，1，2，3

  //根据key获取这个key属于哪一个分区
  //getPartition  每一行数据都会执行一次
  override def getPartition(age: Any): Int = {

    //Math.abs(age.hashCode() % numPartitions)   默认方式，hash分区

    age match {
      case 21 => 0
      case 22 => 1
      case 23 => 2
      case 24 => 3
      case _ => 0
    }
  }
}